package main

import (
	"errors"
	"fmt"
	"github.com/nsqio/go-nsq"
	"os"
	"os/signal"
	"syscall"
	"time"
)

// MyHandler 是一个消费者类型
type MyMessageHandler struct {
	Title string
}

func (m *MyMessageHandler) HandleMessage(msg *nsq.Message) error {
	if len(msg.Body) == 0 {
		// Returning nil will automatically send a FIN command to NSQ to mark the message as processed.
		// In this case, a message with an empty body is simply ignored/discarded.
		return nil
	}
	//处理数据 msg.Body  这里就是简单的打印
	fmt.Printf("%s recv from %v, msg:%v\n", m.Title, msg.NSQDAddress, string(msg.Body))
	return nil

	// Returning a non-nil error will automatically send a REQ command to NSQ to re-queue the message.
	//返回err 会自动让消息重传
	return errors.New("error recv from NSQ")
}
func initConsumers(topic string, channel string, address string) error {
	config := nsq.NewConfig()
	config.LookupdPollInterval = 15 * time.Second
	c, err := nsq.NewConsumer(topic, channel, config)
	if err != nil {
		fmt.Printf("create consumer failed, err:%v\n", err)
		return err
	}
	consumer := &MyMessageHandler{Title: "ybl"}
	c.AddHandler(consumer)
	// if err := c.ConnectToNSQD(address); err != nil { // 直接连NSQD  非集群
	//c.ConnectToNSQLookupds(address []string) // 查询多个不同地址的不同lookup来找相应的 topic  集群方式

	if err := c.ConnectToNSQLookupd(address); err != nil { // 通过单个地址的lookupd查询topic   集群方式
		return err
	}
	return nil
}
func main() {
	err := initConsumers("topic_demo", "first", "127.0.0.1:4161") //单个lookup的默认启动在 4161端口
	if err != nil {
		fmt.Printf("init consumer failed, err:%v\n", err)
		return
	}
	c := make(chan os.Signal)        // 定义一个信号的通道
	signal.Notify(c, syscall.SIGINT) // 转发键盘中断信号到c
	<-c                              // 阻塞
}
